#ifndef __MQ_CHANNEL_HPP__
#define __MQ_CHANNEL_HPP__

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "muduo/proto/dispatcher.h"
#include "../mq_comm/helper.hpp"
#include "../mq_comm/logger.hpp"
#include "../mq_comm/mq_proto.pb.h"
#include "../mq_comm/mq_msg.pb.h"
#include "../mq_comm/mq_threadpool.hpp"
#include "mq_host.hpp"
#include "mq_consumer.hpp"
#include "mq_route.hpp"

namespace mq
{
    using ProtobufCodecPtr=std::shared_ptr<ProtobufCodec>;
    using openChannelRequestPtr=std::shared_ptr<openChannelRequest>;
    using closeChannelRequestPtr=std::shared_ptr<closeChannelRequest>;
    using declareExchangeRequestPtr=std::shared_ptr<declareExchangeRequest>;
    using deleteExchangeRequestPtr=std::shared_ptr<deleteExchangeRequest>;
    using declareQueueRequestPtr=std::shared_ptr<declareQueueRequest>;
    using deleteQueueRequestPtr=std::shared_ptr<deleteQueueRequest>;
    using queueBindRequestPtr=std::shared_ptr<queueBindRequest>;
    using queueUnBindRequestPtr=std::shared_ptr<queueUnBindRequest>;
    using basicPublishRequestPtr=std::shared_ptr<basicPublishRequest>;
    using basicAckRequestPtr=std::shared_ptr<basicAckRequest>;
    using basicConsumeRequestPtr=std::shared_ptr<basicConsumeRequest>;
    using basicCancelRequestPtr=std::shared_ptr<basicCancelRequest>;
        //     std::string _cid;//信道唯一标志ID
        // Consumer::ptr _consumer;//信道关联的消费者句柄
        // muduo::net::TcpConnectionPtr _conn;//链接通信句柄
        // ProtobufCodecPtr _codec;//protobuf协议处理句柄
        // ConsumerManager::ptr _cmp;;//消费者管理句柄
        // VirtualHost::ptr _host;//虚拟机管理句柄
        // ThreadPool::ptr _pool;//线程池操作句柄
    class Channel
    {
    public:
        using ptr=std::shared_ptr<Channel>;
        Channel(const std::string& id,
                const VirtualHost::ptr& host,
                const ConsumerManager::ptr& cmp,
                const ProtobufCodecPtr& codec,
                const muduo::net::TcpConnectionPtr& conn,
                const ThreadPool::ptr& pool):_cid(id),_consumer(nullptr),_conn(conn),_codec(codec),_cmp(cmp),_host(host),_pool(pool)
        {
            DLOG("new Channel: %p",this);
        }
        ~Channel()
        {
            if(_consumer.get()!=nullptr) _cmp->remove(_consumer->tag,_consumer->qname);
            DLOG("del Channel: %p",this);
        }
        /*声明交换机*/
        void declareExchange(const declareExchangeRequestPtr& req)
        {
            bool ret=_host->declareExchange(req->exchange_name(),req->exchange_type(),
                                            req->durable(),req->auto_delete(),req->args());
            return basicResponse(ret,req->rid(),req->cid());
        }

        /*删除交换机*/
        void deleteExchange(const deleteExchangeRequestPtr& req)
        {
            _host->deleteExchange(req->exchange_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        /*声明队列*/
        void declareQueue(const declareQueueRequestPtr& req)
        {
            //虚拟机声明队列
            bool ret=_host->declareQueue(req->queue_name(),req->durable(),req->exclusive(),req->auto_delete(),req->args());
            if(ret==false)
            {
                return basicResponse(false,req->rid(),req->cid());
            }
            //初始化这个队列上的消费者队列
            _cmp->initQueueConsumer(req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        /*删除队列*/
        void deleteQueue(const deleteQueueRequestPtr& req)
        {
            //1.删除队列属性信息和队列本身信息
            _host->deleteQueue(req->queue_name());
            //2.删除订阅了队列的消费者信息
            _cmp->destroyQueueConsumer(req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        /*绑定队列*/
        void queueBind(const queueBindRequestPtr& req)
        {
            bool ret=_host->bind(req->exchange_name(),req->queue_name(),req->binding_key());
            return basicResponse(ret,req->rid(),req->cid());
        }

        /*取消绑定*/
        void queueUnBind(const queueUnBindRequestPtr& req)
        {
            _host->unBind(req->exchange_name(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }

        /*发布消息*/
        void basicPublish(const basicPublishRequestPtr& req)
        {
            //1.确定交换机是否存在
            Exchange::ptr ep=_host->selectExchange(req->exchange_name());
            if(ep.get()==nullptr)
            {
                return basicResponse(false,req->rid(),req->cid());
            }
            //2.获取交换机的绑定队列
            MsgQueueBindingMap exchange_binding=_host->exchangeBinding(req->exchange_name());
            //3.进行路由交换,判断当前消息应该添加进哪一个消息队列中
            BasicProperties* properties=nullptr;
            std::string routing_key;
            if(req->has_properties()==true)
            {
                properties=req->mutable_properties();
                routing_key=properties->routing_key();
            }
            for(auto& binding:exchange_binding)
            {
                if(Router::route(ep->type,routing_key,binding.second->bingding_key)==true)
                {
                    //4.向符合要求的队列中添加消息
                    _host->basicPublish(binding.first,properties,req->body());
                    //5.向异步线程池中添加处理函数
                    auto task=std::bind(&Channel::consume,this,binding.first);
                    _pool->push(task);
                }
            }
            return basicResponse(true,req->rid(),req->cid());
        }

        /*消息的确认*/
        void basicAck(const basicAckRequestPtr& req)
        {
            _host->basicAck(req->queue_name(),req->message_id());
            return basicResponse(true,req->rid(),req->cid());
        }

        /*订阅消息*/
        void basicConsume(const basicConsumeRequestPtr& req)
        {
            //1.判断队列是否存在
            bool ret=_host->existsQueue(req->queue_name());
            if(ret==false)
            {
                return basicResponse(false,req->rid(),req->cid());
            }
            //2.创建消费者
            auto cb=std::bind(&Channel::callback,this,std::placeholders::_1,std::placeholders::_2,std::placeholders::_3);
            _consumer=_cmp->create(req->consumer_tag(),req->queue_name(),req->auto_ack(),cb);
            return basicResponse(true,req->rid(),req->cid());
        }

        /*取消订阅*/
        void basicCancel(const basicCancelRequestPtr& req)
        {
            _cmp->remove(req->consumer_tag(),req->queue_name());
            return basicResponse(true,req->rid(),req->cid());
        }
    private:
        /*回复通用响应*/
        void basicResponse(bool ok,const std::string& rid,const std::string& cid)
        {
            basicCommonResponse resp;
            resp.set_ok(ok);
            resp.set_rid(rid);
            resp.set_cid(cid);
            _codec->send(_conn,resp);
        }

        /*消费者回调函数*/
        void callback(const std::string& tag,const BasicProperties* bp,const std::string& body)
        {
            //针对参数组织出推送消息请求,将消息推送给Channel对应的客户端
            basicConsumeResponse resp;
            resp.set_cid(_cid);
            resp.set_consumer_tag(tag);
            resp.set_body(body);
            if(bp!=nullptr)
            {
                resp.mutable_properties()->set_id(bp->id());
                resp.mutable_properties()->set_delivery_mode(bp->delivery_mode());
                resp.mutable_properties()->set_routing_key(bp->routing_key());
            }
            _codec->send(_conn,resp);
        }

        /*线程池内的线程回调函数*/
        void consume(const std::string& qname)
        {
            //1.从消息队列中获取一个消息
            MessagePtr mp=_host->basicConsume(qname);
            if(mp.get()==nullptr)
            {
                DLOG("执行消费任务失败,%s队列没有消息",qname.c_str());
                return;
            }
            //2.从消费者队列中取出一个消费者
            Consumer::ptr cp=_cmp->choose(qname);
            if(cp.get()==nullptr)
            {
                DLOG("执行消费任务失败,%s队列中没有消费者",qname.c_str());
                return;
            }
            //3.调用消费者的回调处理函数,实现消息的推送
            cp->callback(cp->tag,mp->mutable_payload()->mutable_properties(),mp->payload().body());
            //4.判断是否是自动确认的
            if(cp->auto_ack==true) _host->basicAck(qname,mp->payload().properties().id());
        }
    private:
        std::string _cid;//信道唯一标志ID
        Consumer::ptr _consumer;//信道关联的消费者句柄
        muduo::net::TcpConnectionPtr _conn;//链接通信句柄
        ProtobufCodecPtr _codec;//protobuf协议处理句柄
        ConsumerManager::ptr _cmp;;//消费者管理句柄
        VirtualHost::ptr _host;//虚拟机管理句柄
        ThreadPool::ptr _pool;//线程池操作句柄
    };

    class ChannelManager
    {
    public:
        using ptr=std::shared_ptr<ChannelManager>;
        ChannelManager(){}

        /*创建打开信道*/
        bool openChannel(const std::string& id,
                        const VirtualHost::ptr& host,
                        const ConsumerManager::ptr& cmp,
                        const ProtobufCodecPtr& codec,
                        const muduo::net::TcpConnectionPtr& conn,
                        const ThreadPool::ptr& pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto iter=_channels.find(id);
            if(iter!=_channels.end())
            {
                DLOG("信道: %s已存在",id.c_str());
                return false;
            }
            Channel::ptr channel=std::make_shared<Channel>(id,host,cmp,codec,conn,pool);
            _channels.insert(std::make_pair(id,channel));
            return true;
        }

        /*销毁关闭信道*/
        void closeChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto iter=_channels.find(id);
            if(iter==_channels.end()) return;
            _channels.erase(id);
            return;
        }

        /*根据信道ID获取信道句柄*/
        Channel::ptr getChannel(const std::string& id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto iter =_channels.find(id);
            if(iter==_channels.end()) return Channel::ptr();
            return iter->second;
        }
    private:
        std::mutex _mutex;//互斥锁
        std::unordered_map<std::string,Channel::ptr> _channels;//信道管理哈希
    };
}

#endif